package com.waleipt.api.es;

import java.io.IOException;
import java.net.InetAddress;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import org.elasticsearch.Version;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.cluster.service.ClusterService;
import org.elasticsearch.common.CheckedBiConsumer;
import org.elasticsearch.common.UUIDs;
import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
import org.elasticsearch.common.io.stream.StreamInput;
import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.common.network.NetworkService;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.text.Text;
import org.elasticsearch.common.transport.TransportAddress;
import org.elasticsearch.common.util.BigArrays;
import org.elasticsearch.indices.breaker.CircuitBreakerService;
import org.elasticsearch.indices.breaker.NoneCircuitBreakerService;
import org.elasticsearch.node.Node;
import org.elasticsearch.tasks.TaskManager;
import org.elasticsearch.threadpool.ThreadPool;
import org.elasticsearch.transport.ConnectionProfile;
import org.elasticsearch.transport.RequestHandlerRegistry;
import org.elasticsearch.transport.TcpTransport;
import org.elasticsearch.transport.Transport;
import org.elasticsearch.transport.Transport.Connection;
import org.elasticsearch.transport.TransportException;
import org.elasticsearch.transport.TransportRequest;
import org.elasticsearch.transport.TransportRequestOptions;
import org.elasticsearch.transport.TransportResponse;
import org.elasticsearch.transport.TransportResponseHandler;
import org.elasticsearch.transport.TransportResponseOptions;
import org.elasticsearch.transport.TransportService;
import org.elasticsearch.transport.TransportServiceAdapter;
import org.elasticsearch.transport.TransportSettings;
import org.elasticsearch.transport.netty4.Netty4Transport;
import org.junit.Test;

public class TransportServiceTest {

	public class UserRequest extends TransportRequest
    {
		String ACTION = "foobarAction" ;
		
        int id ;
        String name ;
        int age ;
        
        
        public UserRequest() {}
        
        public UserRequest( int id, String name, int age ) {
            this.id = id ;
            this.name = name ;
            this.age = age ;
            
        }
        
        @Override
        public void readFrom(StreamInput in) throws IOException {
//            super.readFrom(in);
            this.id = in.readInt() ;
            Text text = in.readText() ;
            this.name =  text.toString() ;
            this.age = in.readInt() ;
        }
        
        @Override
        public void writeTo(StreamOutput out) throws IOException {
//            super.writeTo(out);
            out.writeInt( this.id );
            out.writeText( new Text(this.name) );
            out.writeInt( age );
        }
        
    }
    
    public class UserResponse extends TransportResponse
    {
    	String ACTION = "foobarAction" ;
    	
        int status ;
        String reason ;
        
        public UserResponse(){}
        
        public UserResponse( int status, String reason )
        {
            this.status = status ;
            this.reason = reason ;
        }
        
        @Override
        public void readFrom(StreamInput in) throws IOException {
//            super.readFrom(in);
            this.status = in.readInt() ;
            this.reason = in.readString();
        }
        
        @Override
        public void writeTo(StreamOutput out) throws IOException {
//            super.writeTo(out);
            out.writeInt( this.status );
            out.writeString( this.reason );
        }
    }
	
	
    
    /**
     * 自定义协议交互
     * 
     */
    @Test
    public void test_custom_protocol()
    {
        
        ThreadPool threadPool = new ThreadPool(Settings.builder().put(Node.NODE_NAME_SETTING.getKey(), this.getClass().getName() ).build()) ;

        CircuitBreakerService circuitBreakerService = new NoneCircuitBreakerService();
        
        String ACTION_NAME = "walleipt:user" ;
        NamedWriteableRegistry registry = new NamedWriteableRegistry(Collections.emptyList());
        
        Settings settings_node_0 = Settings.builder()
//                .put(TransportSettings.BIND_HOST.getKey(), "0.0.0.0" )
                .put(TransportSettings.BIND_HOST.getKey(), "127.0.0.1,10.100.33.155" )
                .put(TransportSettings.PORT.getKey(), 9300)
            .put(TransportSettings.PUBLISH_HOST.getKey(), "10.100.33.155" )
            .put(TransportSettings.PUBLISH_PORT.getKey(), 9300)
            .put("cluster.name", "test")
            .build();
        final Netty4Transport netty_node_0 = new Netty4Transport(settings_node_0, threadPool, new NetworkService(settings_node_0, Collections.emptyList()),
            BigArrays.NON_RECYCLING_INSTANCE, registry, circuitBreakerService);
        TransportService services_node_0 = new TransportService( settings_node_0, netty_node_0, threadPool, TransportService.NOOP_TRANSPORT_INTERCEPTOR, 
                (boundAddress) -> DiscoveryNode.createLocal( settings_node_0, boundAddress.publishAddress(), settings_node_0.get(Node.NODE_NAME_SETTING.getKey(),
            UUIDs.randomBase64UUID())), null) ;
        
        services_node_0.registerRequestHandler( ACTION_NAME, UserRequest::new, ThreadPool.Names.SAME, (request, channel)->{
	        System.err.println( String.format( "node_0:接收请求参数:id/name/name2/age:%d/%s/%d", request.id, request.name, request.age ) );
	        channel.sendResponse( new UserResponse( 3, "099999") );
        });
        services_node_0.start();
        services_node_0.acceptIncomingRequests();

        Settings settings_node_1 = Settings.builder()
                .put(TransportSettings.BIND_HOST.getKey(), "0.0.0.0" )
                .put(TransportSettings.PORT.getKey(), 9301)
                .put(TransportSettings.PUBLISH_HOST.getKey(), "10.100.33.155" )
                .put(TransportSettings.PUBLISH_PORT.getKey(), 9301)
                .put("cluster.name", "test")
                .build();
        final Netty4Transport netty_node_1 = new Netty4Transport(settings_node_1, threadPool, new NetworkService(settings_node_1, Collections.emptyList()),
            BigArrays.NON_RECYCLING_INSTANCE, registry, circuitBreakerService);
        TransportService services_node_1 = new TransportService( settings_node_1, netty_node_1, threadPool, TransportService.NOOP_TRANSPORT_INTERCEPTOR, 
                (boundAddress) -> DiscoveryNode.createLocal( settings_node_1, boundAddress.publishAddress(), settings_node_1.get(Node.NODE_NAME_SETTING.getKey(),
            UUIDs.randomBase64UUID())), null) ;

            services_node_1.registerRequestHandler( ACTION_NAME, UserRequest::new, ThreadPool.Names.SAME, (request, channel)->{
            System.err.println( String.format( "node_1:接收请求参数:id/name/age:%d/%s/%d", request.id, request.name, request.age ) );
            channel.sendResponse( new UserResponse( 3, "199999") );
        });
        
        services_node_1.start();
        services_node_1.acceptIncomingRequests();

        DiscoveryNode node_0 = services_node_0.getLocalNode();
        DiscoveryNode node_1 = services_node_1.getLocalNode();

        services_node_0.connectToNode(node_1);
        
      
        System.err.println("==========================================================SendRequest==========================================================");
        
        
        String str ="" ;
        for( int i=0; i<254; i++)
        {
            str = str+"1" ;
        }
        
        services_node_0.sendRequest( node_1, "walleipt:user", new UserRequest(3,str,1), new TransportResponseHandler<UserResponse>() {

            @Override
            public UserResponse newInstance() {
                System.err.println( "newInstance" );
                return new UserResponse();
            }

            @Override
            public void handleResponse(UserResponse response) {
                System.err.println( "handleResponse" );
                System.err.println( String.format( "node_0:接收返回参数:status/reason/:%d/%s", response.status, response.reason ) );
            }

            @Override
            public void handleException(TransportException exp) {
                System.err.println( "handleException" );
            }

            @Override
            public String executor() {
                return ThreadPool.Names.SAME;
            }
        });
    }
    
    
    class NoneTransportServiceAdapter implements TransportServiceAdapter
    {

        volatile Map<String, RequestHandlerRegistry> requestHandlers = new HashMap<>() ;
    	
        
        public NoneTransportServiceAdapter( Settings settings) {
        	TaskManager taskManager = new TaskManager(settings) ;
			String executor = ThreadPool.Names.SAME ;
			boolean forceExecution = false ;
			boolean canTripCircuitBreaker = false ;
			RequestHandlerRegistry reg = new RequestHandlerRegistry<>("", UserRequest::new, taskManager, 
											(request, channel)->{
									            System.err.println( String.format( "处理客户端请求参数:id/name/age:%d/%s/%d",  request.id, request.name, request.age ) );
									            channel.sendResponse( new UserResponse( 3, "199999") );
									        }, 
										executor, forceExecution, canTripCircuitBreaker) ;
			requestHandlers.put( "foobarAction", reg) ;
        }
        
		@Override
		public void addBytesReceived(long size) {
			// TODO Auto-generated method stub
			
		}

		@Override
		public void addBytesSent(long size) {
			// TODO Auto-generated method stub
			
		}

		@Override
		public void onRequestSent(DiscoveryNode node, long requestId, String action, TransportRequest request,
				TransportRequestOptions options) {
			// TODO Auto-generated method stub
			
		}

		@Override
		public void onResponseSent(long requestId, String action, TransportResponse response,
				TransportResponseOptions options) {
		}

		@Override
		public void onResponseSent(long requestId, String action, Exception e) {
			// TODO Auto-generated method stub
			
		}

		@Override
		public TransportResponseHandler onResponseReceived(long requestId) {
			return clientHandler.remove( requestId ) ;
		}

		@Override
		public void onRequestReceived(long requestId, String action) {
			// TODO Auto-generated method stub
			
		}

		@Override
		public RequestHandlerRegistry getRequestHandler(String action) {
			return requestHandlers.get(action);
		}
    	
    }
    
    
    Map<Long,TransportResponseHandler> clientHandler = new HashMap<>() ;
    @Test
    public void netty4Transport()
    {
		ThreadPool threadPool = null ;
		try
		{
			
			Settings server_node_settings = Settings.builder()
								.put(TcpTransport.PING_SCHEDULE.getKey(), "1000ms" )
								.put(TransportSettings.BIND_HOST.getKey(), "127.0.0.1" )
								.put(TransportSettings.PUBLISH_HOST.getKey(), "127.0.0.1" )
								.put(TransportSettings.PORT.getKey(), 9300 )
								.build() ;
	        threadPool = new ThreadPool(Settings.builder().put(Node.NODE_NAME_SETTING.getKey(), this.getClass().getName() ).build()) ;
	        
	        
	        Netty4Transport server_node_transport = new Netty4Transport(server_node_settings, 
	        									threadPool, 
	        									new NetworkService(server_node_settings, Collections.emptyList()), 
	        									new BigArrays(Settings.EMPTY, null), 
	        									new NamedWriteableRegistry(Collections.emptyList()), 
	        									new NoneCircuitBreakerService()) ;
	        // 此TransportServiceAdapter会对不同业务协议进行接受处理
	        server_node_transport.transportServiceAdapter( new NoneTransportServiceAdapter(server_node_settings) );
	        
	        // 启动服务端-端口
	        server_node_transport.start();
	        
	        
	        
	        // 客户端链接服务端
	        Settings client_node_settings = Settings.builder()
					.put(TcpTransport.PING_SCHEDULE.getKey(), "1000ms" )
					.put(TransportSettings.BIND_HOST.getKey(), "127.0.0.1" )
					.put(TransportSettings.PUBLISH_HOST.getKey(), "127.0.0.1" )
					.put(TransportSettings.PORT.getKey(), 9301 )
					.build() ;
			Netty4Transport client_node_transport = new Netty4Transport(client_node_settings, 
												threadPool, 
												new NetworkService(client_node_settings, Collections.emptyList()), 
												new BigArrays(Settings.EMPTY, null), 
												new NamedWriteableRegistry(Collections.emptyList()), 
												new NoneCircuitBreakerService()) ;
			// 此TransportServiceAdapter会对不同业务协议进行接受处理
			client_node_transport.transportServiceAdapter( new NoneTransportServiceAdapter(client_node_settings) );
			
			// 启动服务端-端口
			client_node_transport.start();
	        UserRequest request = new UserRequest( 1, "name", 23) ;
			String nodeId = "foo" ;
			String address = "127.0.0.1" ;
			int port = 9300 ;
			DiscoveryNode node = new DiscoveryNode( nodeId, new TransportAddress( InetAddress.getByName(address), port), Version.CURRENT);
			// 连接Node,发送连接node请求报文
			client_node_transport.connectToNode(node, null, new CheckedBiConsumer<Transport.Connection, ConnectionProfile, IOException>() {
				
				@Override
				public void accept(Connection t, ConnectionProfile u) throws IOException {
					
				}
			} );
	        Transport.Connection connection = client_node_transport.getConnection(node);
	        String action = "foobarAction" ;
	        long requestId = 42 ;
	        connection.sendRequest( requestId, action, request, TransportRequestOptions.EMPTY );
	        clientHandler.put( requestId, new TransportResponseHandler<UserResponse>() {
	            @Override
	            public UserResponse newInstance() {
	                return new UserResponse();
	            }

	            @Override
	            public void handleResponse(UserResponse response) {
	                System.err.println( String.format( "处理服务端返回结果:status/reason/:%d/%s", response.status, response.reason ) );
	            }

	            @Override
	            public void handleException(TransportException exp) {
	            }

	            @Override
	            public String executor() {
	                return ThreadPool.Names.SAME;
	            }
	        }) ;
	        
	        
	        Thread.sleep( 5000 );
	        
		}
		catch( Throwable err )
		{
			err.printStackTrace();
		}
		finally 
		{
	        ThreadPool.terminate(threadPool, 10, TimeUnit.SECONDS);
	    }
    }
    
	
	
	@Test 
	public void testTransportService()
	{
		
	}
	
}
